package com.k.activemq.openwire2;

import com.k.activemq.ActiveMQUtils;
import com.k.activemq.MqConnectionEntity;

import javax.jms.*;
import java.nio.charset.StandardCharsets;

public class MqPublisher {
    public static void main(String[] args) {
//        MqConnectionEntity connectionEntity = ActiveMQUtils.getConnectionData();
//        MqConnectionEntity connectionEntity = MqConnectionEntity.newBuilder()
//                .setUrl("tcp://localhost:61616")  // brokerURL
//                .setUsername("admin")             // username
//                .setPassword("admin")             // password
//                .setDestination("ljw-queue")      // Topic or Queue Name
//                .setPubSubDomain(false)            // true: Topic
//                .setMessage("hello")              // Message
//                .setByteMessage(true)             // true: ByteMessage
//                .build();
        MqConnectionEntity connectionEntity = MqConnectionEntity.newBuilder()
                .setUrl("tcp://192.168.241.89:22000")  // brokerURL
                .setUsername("admin")             // username
                .setPassword("Command_123")             // password
                .setDestination("start_report")      // Topic or Queue Name
                .setPubSubDomain(true)            // true: Topic
                .setMessage("hello")              // Message
                .setByteMessage(true)             // true: ByteMessage
                .build();
        System.out.println("mq publisher:"+connectionEntity.toString());
        try {
            System.out.println("Mq start publish ..." + connectionEntity.getPubSubDomain() + ":" + connectionEntity.getDestination());
            publish(connectionEntity);
            System.out.println("Mq start publish ok");
        } catch (JMSException e) {
            System.out.println("Mq start publish Error");
            e.printStackTrace();
        }
        System.out.println("Mq start finished!");
    }

    private static void publish(MqConnectionEntity connEntity) throws JMSException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = ActiveMQUtils.getActiveMQConnectionFactory(connEntity.getUrl());
        Connection connection = connectionFactory.createConnection(connEntity.getUsername(), connEntity.getPassword());
        // 创建连接
//        Connection connection = connectionFactory.createConnection();
        // 打开连接
        connection.start();
        // 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建队列目标,并标识队列名称，消费者根据队列名称接收数据
        Destination destination = resolveDestinationName(connEntity, session);
        // 创建一个生产者
        MessageProducer producer = session.createProducer(destination);
        // 向队列推送消息数据
        Message message = getMessage(connEntity, session);
        //发送消息
        producer.send(message);
        //关闭连接
        connection.close();
    }

    private static Destination resolveDestinationName(MqConnectionEntity entity, Session session) throws JMSException {
        return (Destination)(entity.getPubSubDomain() ? resolveTopic(session, entity.getDestination()) : resolveQueue(session, entity.getDestination()));
    }
    protected static Topic resolveTopic(Session session, String topicName) throws JMSException {
        return session.createTopic(topicName);
    }

    protected static Queue resolveQueue(Session session, String queueName) throws JMSException {
        return session.createQueue(queueName);
    }

    private static Message getMessage(MqConnectionEntity entity, Session session) throws JMSException {
        if (entity.getByteMessage()) {
            BytesMessage message = session.createBytesMessage();
            byte[] byteMsg = entity.getMessage().getBytes(StandardCharsets.UTF_8);
            message.writeBytes(byteMsg);
            return message;
        } else {
            return session.createTextMessage(entity.getMessage());
        }
    }
}
